/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * openGauss is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 * http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * ---------------------------------------------------------------------------------------
 *
 * imcu_storage.cpp
 * routines to support ColStore
 *
 * IDENTIFICATION
 * src/gausskernel/storage/imcstore/imcu_storage.cpp
 *
 * ---------------------------------------------------------------------------------------
 */

#include "postgres.h"
#include "knl/knl_variable.h"
#include <fcntl.h>
#include <sys/file.h>
#include "miscadmin.h"
#include "access/heapam.h"
#include "access/genam.h"
#include "access/xlog.h"
#include "catalog/pg_tablespace.h"
#include "catalog/catalog.h"
#include "commands/tablespace.h"
#include "service/remote_read_client.h"
#include "access/imcs/imcs_allocspace.h"
#include "access/imcs/imcu.h"
#include "access/imcs/imcu_storage.h"
#include "access/imcs/imcu_cache_mgr.h"
#include "access/imcs/imcs_ctlg.h"
#include "storage/lmgr.h"
#include "pgxc/pgxc.h"
#include "postmaster/postmaster.h"
#include "utils/aiomem.h"
#include "utils/plog.h"
#include "securec_check.h"

IMCUStorage::IMCUStorage(const CFileNode &cfile_node, CStoreAllocateStrategy strategy) : CUStorage(cfile_node, strategy)
{
    InitFileNamePrefix(cfile_node);
}

void IMCUStorage::Destroy()
{
    if (m_fd != FILE_INVALID)
        CloseFile(m_fd);

    if (m_freespace != NULL) {
        DELETE_EX(m_freespace);
    }
}

IMCUStorage::~IMCUStorage()
{
    if (m_freespace != NULL) {
        DELETE_EX(m_freespace);
    }
}

FORCE_INLINE
void IMCUDesc::Reset()
{
    cu_id = InValidCUID;
    errno_t rc = 0;
    rc = memset_s(cu_min, MIN_MAX_LEN, 0, MIN_MAX_LEN);
    securec_check(rc, "\0", "\0");
    rc = memset_s(cu_max, MIN_MAX_LEN, 0, MIN_MAX_LEN);
    securec_check(rc, "\0", "\0");
    row_count = 0;
    cu_size = 0;
    cu_mode = 0;
    cu_pointer = 0;
    magic = 0;
    xmin = 0;
}

IMCUDesc::IMCUDesc()
{
    Reset();
}

IMCUDesc::~IMCUDesc() {}

IMCSDesc::~IMCSDesc()
{
    if (deletedList != NULL) {
        for (uint32 rg_idx = 0; rg_idx < rowgroup_num; ++rg_idx) {
            if (deletedList[rg_idx] != NULL) {
                DELETE_EX(deletedList[rg_idx]);
            }
        }
        pfree(deletedList);
    }
}

int IMCSDesc::markDeleted(uint32 rg_idx, uint32 row_idx)
{
    MemoryContext oldcontext = NULL;
    Assert(rg_idx < rowgroup_num);

    if (deletedList == NULL) {
        oldcontext = MemoryContextSwitchTo(IMCU_CACHE->m_imcs_context);
        LWLockAcquire(imcu_desc_latch, LW_EXCLUSIVE);
        deletedList = (htap::HTAPDeleteMap **)palloc0(sizeof(htap::HTAPDeleteMap *) * rowgroup_num);
        LWLockRelease(imcu_desc_latch);
        (void)MemoryContextSwitchTo(oldcontext);
    }
    if (deletedList[rg_idx] == NULL) {
        oldcontext = MemoryContextSwitchTo(IMCU_CACHE->m_imcs_context);
        LWLockAcquire(imcu_desc_latch, LW_EXCLUSIVE);
        deletedList[rg_idx] = New(IMCU_CACHE->m_imcs_context) htap::HTAPDeleteMap();
        deletedList[rg_idx]->init(IMCS_MAX_ROWS_SIZE);
        LWLockRelease(imcu_desc_latch);
        (void)MemoryContextSwitchTo(oldcontext);
    }
    return deletedList[rg_idx]->set(row_idx);
}

bool IMCSDesc::isDeleted(uint32 rg_idx, uint32 row_idx)
{
    Assert(rg_idx < rowgroup_num);

    return (deletedList && (deletedList[rg_idx]) && deletedList[rg_idx]->check(row_idx));
}

int IMCSDesc::increaseRGNum()
{
    if (deletedList != NULL) {
        MemoryContext oldcontext = MemoryContextSwitchTo(IMCU_CACHE->m_imcs_context);
        htap::HTAPDeleteMap **deleteList = (htap::HTAPDeleteMap **)palloc(sizeof(htap::HTAPDeleteMap *) * (rowgroup_num + 1));
        (void)MemoryContextSwitchTo(oldcontext);

        memcpy(deleteList, deletedList, sizeof(htap::HTAPDeleteMap *) * rowgroup_num);
        deleteList[rowgroup_num] = NULL;

        htap::HTAPDeleteMap **saveDeleteList = deletedList;
        LWLockAcquire(imcu_desc_latch, LW_EXCLUSIVE);
        deletedList = deleteList;
        LWLockRelease(imcu_desc_latch);
        pfree(saveDeleteList);
    }
    IMCU_CACHE->CreateImcsRgLock(imcs_oid, rowgroup_num);
    rowgroup_num++;
    return 0;
}


static void SaveCUReportIOError(const char *file_name, uint64 writeOffset, int writtenBtyes, int expectedBytes,
    int expectedTotalBytes, int align_size)
{
    if (writtenBtyes > 0) {
        Assert(align_size > 0);
        if (writtenBtyes % align_size != 0) {
            /* later load will report error about not-matched align */
            ereport(ERROR, (errcode_for_file_access(),
                errmsg("CU partial write, file \"%s\", offset(%lu), total(%d), expected(%d), actual(%d): %m", file_name,
                writeOffset, expectedTotalBytes, expectedBytes, writtenBtyes),
                errhint("CU file will be not page aligned, Check free disk space")));
        } else {
            ereport(ERROR, (errcode_for_file_access(),
                errmsg("CU partial write, file \"%s\", offset(%lu), total(%d), expected(%d), actual(%d): %m", file_name,
                writeOffset, expectedTotalBytes, expectedBytes, writtenBtyes),
                errhint("Check free disk space.")));
        }
    } else {
        ereport(ERROR, (errcode_for_file_access(),
            errmsg("could not extend file \"%s\", offset(%lu), total(%d), expected(%d), actual(%d): %m", file_name,
            writeOffset, expectedTotalBytes, expectedBytes, writtenBtyes),
            errhint("Check free disk space.")));
    }
}

static void LoadCUReportIOError(const char *file_name, uint64 read_offset, int read_bytes, int expectedBytes,
    int expectedTotalBytes)
{
    if (errno == ENOENT) {
        /* file not exists */
        ereport(ERROR, (errcode_for_file_access(), errmsg("read file \"%s\" failed, %m", file_name),
            errdetail("offset(%lu), load size(%d), expected read(%d), actual read(%d)", read_offset,
                                  expectedTotalBytes, expectedBytes, read_bytes)));
    } else if (read_bytes >= 0) {
        if (read_bytes == 0) {
            /* no data is read from */
            ereport(ERROR, (errcode_for_file_access(),
                errmsg("no data is read from \"%s\", offset(%lu), load size(%d), expected read(%d), actual read(%d)",
                file_name, read_offset, expectedTotalBytes, expectedBytes, read_bytes)));
        } else {
            /* only some data is read from */
            ereport(ERROR, (errcode_for_file_access(),
                errmsg("some data are read from \"%s\", offset(%lu), load size(%d), expected read(%d), actual "
                "read(%d)",
                file_name, read_offset, expectedTotalBytes, expectedBytes, read_bytes),
                errdetail("maybe CU file is not page aligned, or with not-completed CU data"),
                errhint("maybe you should upgrade cstore data files first, or there were CU written IO error before")));
        }
    } else {
        ereport(ERROR, (errcode_for_file_access(),
            errmsg("could not read file \"%s\", offset(%lu), load size(%d), expected read(%d), actual read(%d): %m",
            file_name, read_offset, expectedTotalBytes, expectedBytes, read_bytes)));
    }
}

static inline int min(int a, int b)
{
    return (a > b ? b : a);
}

void IMCUStorage::InitFileNamePrefix(_in_ const CFileNode &cfile_node)
{
    int pathlen;
    char attr_name[32];
    errno_t rc = EOK;

    m_fileNamePrefix[0] = '\0'; /* empty string */

    rc = sprintf_s(attr_name, sizeof(attr_name), "C%d", cfile_node.m_attid);
    securec_check_ss(rc, "", "");

    Oid spcoid, dboid, reloid;
    spcoid = cfile_node.m_rnode.spcNode;
    dboid = cfile_node.m_rnode.dbNode;
    reloid = cfile_node.m_rnode.relNode;

    if (spcoid == GLOBALTABLESPACE_OID) {
        /* Shared system relations live in {datadir}/global */
        Assert(dboid == 0);
        pathlen = strlen("global/imcu") + 1 + OIDCHARS + 1 + strlen(attr_name) + 1;
        rc = snprintf_s(m_fileNamePrefix, sizeof(m_fileNamePrefix), pathlen, "global/imcu/%u_%s", reloid, attr_name);
        securec_check_ss(rc, "", "");
    } else if (spcoid == DEFAULTTABLESPACE_OID) {
        /* The default tablespace is {datadir}/base */
        pathlen = strlen("base/imcu") + 1 + OIDCHARS + 1 + OIDCHARS + 1 + strlen(attr_name) + 1;
        rc = snprintf_s(m_fileNamePrefix, sizeof(m_fileNamePrefix), pathlen, "base/imcu/%u_%u_%s", dboid, reloid,
            attr_name);
        securec_check_ss(rc, "", "");
    } else {
        /* All other tablespaces are accessed via symlinks */
        rc = snprintf_s(m_fileNamePrefix, sizeof(m_fileNamePrefix), sizeof(m_fileNamePrefix) - 1,
            "pg_tblspc/%u/%s_%s/%u/%u_%s", spcoid, TABLESPACE_VERSION_DIRECTORY,
            g_instance.attr.attr_common.PGXCNodeName, dboid, reloid, attr_name);
        securec_check_ss(rc, "", "");
    }
}

void IMCUStorage::SaveIMCU(_in_ char *write_buf, _in_ int size, _in_ int cu_id, bool direct_flag, bool for_extension)
{
    char tmp_file_name[MAXPGPATH] = {0};
    GetFileName(tmp_file_name, MAXPGPATH, cu_id);

    File tmp_fd = OpenFile(tmp_file_name, cu_id, direct_flag);
    Assert(tmp_fd != FILE_INVALID);

    int writtenBytes = FilePWrite(tmp_fd, write_buf, size, 0);
    if (writtenBytes != size) {
        int align_size = is_2byte_align ? ALIGNOF_TIMESERIES_CUSIZE : ALIGNOF_CUSIZE;
        SaveCUReportIOError(tmp_file_name, 0, writtenBytes, size, size, align_size);
    }
    CloseFile(tmp_fd);
}

void IMCUStorage::Load_data(_in_ int size, __inout char *outbuf, _in_ int cu_id, bool direct_flag)
{
    char tmp_file_name[MAXPGPATH] = {0};
    GetFileName(tmp_file_name, MAXPGPATH, cu_id);
    char *read_buf = outbuf;

    File tmp_fd = OpenFile(tmp_file_name, cu_id, direct_flag);
    Assert(tmp_fd != FILE_INVALID);

    int nbytes = FilePRead(tmp_fd, read_buf, size, 0);
    if (nbytes != size) {
        LoadCUReportIOError(tmp_file_name, 0, nbytes, size, size);
    }
    CloseFile(tmp_fd);
    remove(tmp_file_name);
}

/*
 * @Description: load IMCU data from IMCU file
 * @Param[IN/OUT] cu_ptr: IMCU object to load data
 * @Param[IN] direct_flag: if ADIO feature is enable, DIO is used.
 * @Param[IN] in_cu_cache: whether this cu_ptr is in IMCU cache.
 * @Param[IN] offset: IMCU data logic offset in logic file
 * @Param[IN] size: IMCU data size
 * @See also:
 */
void IMCUStorage::LoadIMCU(_in_ IMCU *cu_ptr, _in_ int cu_id, _in_ int size, bool direct_flag, bool in_cu_cache)
{
    if (size == 0) {
        cu_ptr->m_compressedBufSize = 0;
        ereport(ERROR, (errcode(ERRCODE_DATA_EXCEPTION), errmsg("CUStorage::LoadIMCU size = 0")));
        return;
    }

    if (!IsDataFileExist(cu_id)) {
        char tmp_file_name[MAXPGPATH] = {0};
        GetFileName(tmp_file_name, MAXPGPATH, cu_id);
        ereport(ERROR, (errcode_for_file_access(), errmsg("Could not open file \"%s\": %m", tmp_file_name)));
        return;
    }
    cu_ptr->m_compressedBuf = (char *)IMCSMemAlloc::Palloc(size, !in_cu_cache);
    Load_data(size, cu_ptr->m_compressedBuf, cu_id, direct_flag);


    // the complete IMCU data has been loaded, so set the cu size.
    // we will check this value during decompressing cu data.
    //
    cu_ptr->SetCUSize(size);
    cu_ptr->m_compressedBufSize = size;
    cu_ptr->m_cache_compressed = true;
}

void IMCUStorage::RemoveImcuFile(_in_ int col, _in_ int cu_id)
{
    char tmp_file_name[MAXPGPATH] = {0};
    m_cnode.m_attid = col;
    InitFileNamePrefix(m_cnode);

    if (!IsDataFileExist(cu_id)) {
        GetFileName(tmp_file_name, MAXPGPATH, cu_id);
        ereport(ERROR, (errcode_for_file_access(), errmsg("Could not open file \"%s\": %m", tmp_file_name)));
        return;
    }

    GetFileName(tmp_file_name, MAXPGPATH, cu_id);
    remove(tmp_file_name);
}

/* open file, if file not exists then create it */
File IMCUStorage::WSOpenFile(char *file_name, int file_id, bool direct_flag)
{
    uint32 file_flags = O_RDWR | PG_BINARY;

    struct stat st;

    /* file may create by other, so  file_flags without O_EXCL */
    if (lstat((const char *)file_name, &st) == -1) {
        file_flags = O_RDWR | O_CREAT | PG_BINARY;
    }

    ADIO_RUN()
    {
        if (direct_flag) {
            file_flags |= O_DIRECT;
        }
    }
    ADIO_END();

    RelFileNodeForkNum filenode;
    filenode.rnode.node = m_cnode.m_rnode;
    filenode.rnode.backend = InvalidBackendId;
    filenode.segno = file_id;
    filenode.storage = COLUMN_STORE;
    if (m_cnode.m_attid < 0) {
        ereport(ERROR, (errcode(ERRCODE_INVALID_ATTRIBUTE), errmsg("validate user defined attribute failed!")));
    }
    filenode.forknumber = (ForkNumber)(m_cnode.m_attid + FirstColForkNum);

    return DataFileIdOpenFile(file_name, filenode, (int)file_flags, 0600);
}

IMCUFile::IMCUFile(const RelFileNode &fileNode, int col) : CUFile(fileNode, col) {}

IMCUFile::~IMCUFile()
{
    m_col_storage = NULL;
    m_buffer = NULL;
}

void IMCUFile::Destroy()
{
    DELETE_EX(m_col_storage);

    ADIO_RUN()
    {
        adio_align_free(m_buffer);
    }
    ADIO_ELSE()
    {
        pfree(m_buffer);
        m_buffer = NULL;
    }
    ADIO_END();
}
